1. Descarga y preparación del corpus de datos

El objetivo de este primer notebook es realizar la carga del dataset para el posterior estudio y entrenamiento de modelos.

Importaciones

import librosa
import wave
import IPython.display as ipd

import numpy as np

import matplotlib.pyplot as plt

import datasets
import pandas as pd
from tqdm.auto import tqdm
from pprint import pprint

import os

---

1.1. Carga del conjunto de datos de HuggingFace

En esta sección vamos a presentar el principal conjunto de datos que utilizaremos para la realización de este proyecto, el conjunto de datos CCMUSIC.

Descarga en local del dataset desde HuggingFace

El principal dataset que utilizaremos, CCMUSIC, está disponible en HuggingFace. Al acceder a la página, nos encontramos con una tarjeta que nos proporciona detalles sobre el dataset y cómo obtener los datos.

Figura: CCMUSIC card en HuggingFace

Se nos indica la utilización del paquete datasets de HuggingFace, en concreto el método load_datasets, permitiendo la carga del conjunto de datos desde el repositorio de HuggingFace hacia una variable en nuestro código local.

ccmusic_corpus = datasets.load_dataset("ccmusic-database/music_genre", name="default",trust_remote_code=True)

Observación: la celda anterior tarda bastante tiempo en descargar, debido al elevado tamaño del dataset (aproximadamente 4 horas), por lo que se recomienda no ejecutarla a menos que cuente con tiempo suficiente. Alternativamente, puede proceder a visualizar las salidas del presente notebook ya ejecutadas.

Características generales del conjunto de datos

Estructura del conjunto de datos: campos y particiones

• El conjunto de datos ccmusic_corpus es un DatasetDict, un dataset con estructura de diccionario que contiene tres particiones (train, validation y test). Cuenta con un total de 1700 piezas ditribuidas en 1370 para entrenamiento, 171 para validación y 172 para test.

• Cada una de estas particiones es un objeto de tipo Dataset que tiene los siguientes campos:

  • audio: Este campo contiene información sobre el archivo de audio.
    • path: La ruta al archivo de audio en el sistema de archivos.
    • array: Un array NumPy que representa los datos del archivo de audio.
    • sampling_rate: La frecuencia de muestreo del archivo de audio.
  • mel: Una imagen de las características mel-espectrograma. Es un objeto PIL.JpegImagePlugin.JpegImageFile.
  • fst_level_label: Una etiqueta de clasificación de primer nivel.
  • sec_level_label: Una etiqueta de clasificación de segundo nivel.
  • thr_level_label: Una etiqueta de clasificación de tercer nivel.

  Nosotros tenemos interés en array y fst_level_label.

# Visualizamos la estructura de una entrada
print(ccmusic_corpus)

DatasetDict({
    train: Dataset({
        features: ['audio', 'mel', 'fst_level_label', 'sec_level_label', 'thr_level_label'],
        num_rows: 1370
    })
    validation: Dataset({
        features: ['audio', 'mel', 'fst_level_label', 'sec_level_label', 'thr_level_label'],
        num_rows: 171
    })
    test: Dataset({
        features: ['audio', 'mel', 'fst_level_label', 'sec_level_label', 'thr_level_label'],
        num_rows: 172
    })
})

# Visualizamos la estructura de una entrada del dataset
pprint(str(ccmusic_corpus['train'][0]))

("{'audio': {'path': "
 "'C:\\\\Users\\\\Usuario\\\\.cache\\\\huggingface\\\\datasets\\\\downloads\\\\extracted\\\\9978c1aa27e41c465d1a0a120f523eae02b367b680fff33f401cee81e10d28c4\\\\audio\\\\2_non-classic\\\\11_rock\\\\21_Soft "
 "Rock\\\\6573efb8158239c2015abeaea6bb8f65.mp3', 'array': array([0., 0., 0., "
 "..., 0., 0., 0.]), 'sampling_rate': 22050}, 'mel': "
 '<PIL.JpegImagePlugin.JpegImageFile image mode=RGB size=496x369 at '
 "0x1FD640CE450>, 'fst_level_label': 1, 'sec_level_label': 8, "
 "'thr_level_label': 14}")

Dataset orientado a clasificación

• Se trata de un dataset orientado especialmente a tareas de clasificación. Las piezas musicales están etiquetadas de acuerdo con tres posibles jerarquías:

  1. fst_level_label: separa las piezas de acuerdo a una clasificación binaria en música clásica y no clásica.
  2. sec_level_label: amplía las categorías, permitiendo distinguir hasta 9 géneros músicales dentro de estas 2 categorías principales.
  3. thr_level_label: amplía la jerarquía anterior, permitiendo disitinguir hasta 16 géneros musicales.

• A continuación, se muestra un desglose de las categorías presentes en el dataset:

  • Classic

    • Symphony
    • Opera
    • Solo
    • Chamber

  • Non_classic

    • Pop

      • Pop_vocal_ballad
      • Adult_contemporary
      • Teen_pop

    • Dance_and_house

      • Contemporary_dance_pop
      • Dance_pop

    • Indie

      • Classic_indie_pop
      • Chamber_cabaret_and_art_pop

    • Soul_or_r_and_b

    • Rock

      • Adult_alternative_rock
      • Uplifting_anthemic_rock
      • Soft_rock
      • Acoustic_pop

  En este proyecto trabajaremos fundamentalmente con esta jerarquía de clasificación fst_level_label.

  Vamos a visualizar, por ejemplo, la distribución de las categorías en train. Podemos observar que el dataset no se encuentra balanceado para ninguna de las jerarquías.

partition = "train"

print(f"---------Recuento de etiquetas en {partition} -------------------- ")

for column_name in ['fst_level_label', 'sec_level_label', 'thr_level_label']:
    recuentos = pd.Series(ccmusic_corpus[partition][column_name]).value_counts().sort_index()
    print(f"Jerarquía: {column_name}")	
    print(recuentos)
    print()

---------Recuento de etiquetas en train -------------------- 
Jerarquía: fst_level_label
0     329
1    1041
Name: count, dtype: int64

Jerarquía: sec_level_label
0     84
1     85
2     81
3     79
4    258
5    153
6    164
7    167
8    299
Name: count, dtype: int64

Jerarquía: thr_level_label
0      84
1      85
2      81
3      79
4      89
5      84
6      85
7      76
8      77
9      83
10     81
11    167
12     72
13     76
14     78
15     73
Name: count, dtype: int64

Características de las señales de audio

• Cada pieza está representada por una señal de audio mono, con una tasa de muestreo homogénea de 22050 Hz y una duración variable.

  Vamos a visualizar un par de entradas del dataset de entrenamiento y mostraremos la reproducción del audio, la duración, la tasa de muestreo y su categoría.

index1 = 653
audio1 = ccmusic_corpus['train'][index1]['audio']['array']
sr1 = ccmusic_corpus['train'][index1]['audio']['sampling_rate']
label1 = ccmusic_corpus['train'][index1]['fst_level_label']

print("Audio escogido de train con índice", index1)

plt.figure(figsize=(5,2))
librosa.display.waveshow(audio1,sr=sr1,color="#f44",alpha=0.8)
plt.title(f"Señal de audio de entrenamiento con índice {index1}")
plt.show()

ipd.display(ipd.Audio(audio1, rate=sr1))

print(f"Dimensiones de la señal: {audio1.shape}")
print(f"Duración: {round(len(audio1)/sr1, 3)} segundos")
print(f"Tasa de muestreo: {sr1}")
print(f"Etiqueta (jerarquía 1): {"Classic" if label1 == "Classical" else "Non_classic"}")

Audio escogido de train con índice 653

Dimensiones de la señal: (7735704,)
Duración: 350.826 segundos
Tasa de muestreo: 22050
Etiqueta (jerarquía 1): Non_classic

index2 = 22
audio2 = ccmusic_corpus["train"][index2]["audio"]["array"]
sr2 = ccmusic_corpus["train"][index2]["audio"]["sampling_rate"]
label2 = ccmusic_corpus["train"][index2]["fst_level_label"]

print("Audio escogido de train con índice", index2)

plt.figure(figsize=(5,2))
librosa.display.waveshow(audio2,sr=sr2,color="#44f",alpha=0.8)
plt.title(f"Señal de audio de entrenamiento con índice {index2}")
plt.show()

ipd.display(ipd.Audio(audio2, rate=sr2))

print(f"Dimensiones de la señal: {audio2.shape}")
print(f"Duración: {round(len(audio2)/sr2, 3)} segundos")
print(f"Tasa de muestreo: {sr2}")
print(f"Etiqueta (jerarquía 1): {"Classic" if label2 == "Classical" else "Non_classic"}")

Audio escogido de train con índice 22

Dimensiones de la señal: (2787182,)
Duración: 126.403 segundos
Tasa de muestreo: 22050
Etiqueta (jerarquía 1): Non_classic

# Comprobar que la tasa de muestreo siempre es 22050 y que todas las señales son mono
# Como ejemplo lo vamos a hacer en el conjunto de test 

all_mono_and_sr_22050 = all([ 
    len(audio["audio"]["array"].shape) == 1 and audio["audio"]["sampling_rate"] == 22050 
        for audio in ccmusic_corpus["test"]
   ])

print(f"Todos los audios son mono y tienen tasa de muestreo 22050: {all_mono_and_sr_22050}")

Todos los audios son mono y tienen tasa de muestreo 22050: True

---

1.2. Generación de la estructura de directorios del corpus

En esta sección, queremos extraer del objeto Dataset anterior únicamente la información relevante para nuestro estudio. Nuestro objetivo es crear una estructura de directorios que almacene toda la información relevante del corpus, con el fin de utilizarla para el entrenamiento del modelo.

Nuestro objetivo es obtener una estructura de directorios de corpus con el siguiente formato:

/ccmusic
    /train 
        /audios
            audio_train_1.wav
            ....
            audio_train_1369.wav
        annotations.csv
    /validation
        /audios
        annotations.csv
    /test
        /audios
        annotations.csv

En esta estructura, cada carpeta almacena los datos de cada partición (entrenamiento, validación y prueba). Dentro de cada carpeta, tendremos tres elementos:

1. /audios: Carpeta que almacena todos los archivos de audio.
2. annotations.csv: Contiene las anotaciones que asocian cada audio con su categoría correspondiente (audio_file, labelID, labelName).

Definición de funciones auxiliares

A continuación, vamos a proceder a almacenar los ficheros siguiento la estructura de directorios antes explicada. Por limpieza y organización, crearemos una serie de funciones auxiliares que nos permitan desglosar esta funcionalidad en 3 acciones: guardar ficheros de audio, almacenar anotaciones que asocien cada fichero con su categoría y transformar el audio para homogeneizar sus características de cada al preprocesamiento.

• Funciones auxiliares para almacenar el fichero wav y las anotaciones:
  • Almacenar los ficheros WAV a partir de las señales de audio: la función store_audio_file toma un array de audio audio_array y un nombre de archivo audio_file. El archivo se guarda en formato WAV con 1 canal (mono), 16 bits por muestra, y una frecuencia de muestreo de 22050 Hz.
  • Crear fichero de anotaciones: la función store_annotation permite tener un registro de a qué categoría pertenece cada señal de audio. Para almacenar tanto la etiqueta como su nombre/significado, deberemos hacer uso de diccionarios de mapeo.

def store_audio_file(audio_array, audio_file):

    ww_obj=wave.open(audio_file,'w')
    ww_obj.setnchannels(1)
    ww_obj.setsampwidth(2)
    ww_obj.setframerate(22050)

    signal=np.int16(audio_array * 32767)
    ww_obj.writeframesraw(signal)

def store_annotation(label_id, label_name, audio_file, annotations_file):

    with open(annotations_file, "a") as f:
        f.write(f"{audio_file},{label_id},{label_name}\n")

label_mapper_fst_level = {
    0: "Classic",
    1: "Non_classic"
}

label_mapper_sec_level = {
    0: "Symphony",
    1: "Opera",
    2: "Solo",
    3: "Chamber",
    4: "Pop",
    5: "Dance_and_house",
    6: "Indie",
    7: "Soul_or_r_and_b",
    8: "Rock"
}

• Funciones auxiliares para las transformaciones de las señales de audio:
  • Transformar todas las señales a formato mono: no será necesario, ya que actualmente todas las señales están en formato mono.
  • Unificar la tasa de muestreo: no será necesario, ya que todas las señales poseen sr=22050 Hz.
  • Unificar el número de muestras:
    • La fórmula para calcular el número de muestras es:
      $\text{Número de Muestras} = \text{Tasa de Muestreo en Hz} \times \text{Duración en Segundos}$
    • Será necesario unificar el número de muestras, ya que las señales tienen la misma tasa de muestreo, pero diferente duración. En nuestro caso, la tasa de muestreo es 22050 Hz y queremos que la duración sea de 20 segundos. Por lo tanto, el número de muestras será:
      $\text{Número de Muestras} = 22050 \text{ Hz} \times 30 \text{ s} = 661500 \text{ muestras}$

# Función auxiliar para unificar el número de muestras (por ende unificamos la duración al tener todas las señales la misma tasa de muestreo)

NUM_SAMPLES = 22050 * 30                                            # num_mustras = tasa de muestreo (22050 Hz) * duracion en segundos (30 s) = 661500

def resize_signal(signal, num_samples=NUM_SAMPLES):
    if signal.shape[0] > num_samples:                               # Recortar si tiene más muestras
        signal = signal[:num_samples]
    elif signal.shape[0] < num_samples:                             # Rellenar si tiene menos muestras  
        signal = np.pad(signal, (0, num_samples - signal.shape[0]))
    return signal

# Probar la función con un ejemplo

# Una señal con más muestras (recorte)
signal = np.random.randn(800000)
resized_signal = resize_signal(signal)

print(f"Recorte - Número de muestras antes: {signal.shape[0]}")
print(f"Recorte - Número de muestras después: {resized_signal.shape[0]}")

# Una señal con menos muestras (relleno)
signal = np.random.randn(100000)
resized_signal = resize_signal(signal)

print(f"Relleno - Número de muestras antes: {signal.shape[0]}")
print(f"Relleno - Número de muestras después: {resized_signal.shape[0]}")

Recorte - Número de muestras antes: 800000
Recorte - Número de muestras después: 661500
Relleno - Número de muestras antes: 100000
Relleno - Número de muestras después: 661500

Generación de la estructura de directorios del corpus

La siguiente función, generate_corpus_files, nos permitirá organizar el corpus de datos en una estructura de directorios con los ficheros necesarios, obteniéndose una estructura como la siguiente.

Para ello, irá recorriendo las distintas particiones del dataset y para cada audio recortará la señal, guardará el audio y generará la anotación (haciendo uso de las funciones auxiliares antes definidas resize_signal, store_audio_file, store_annotation).

def generate_corpus_files(corpus_data, label_column, label_mapper, corpus_folder):

    # Crear directorio principal
    os.makedirs(corpus_folder, exist_ok=True)

    # Almacenar ficheros de cada partición
    for partition in ["train", "validation", "test"]:

        # Crear subdirectorios para almacenar los audios
        os.makedirs(f"{corpus_folder}/{partition}/audios", exist_ok=True)

        # Crear fichero de anotaciones
        with open(f"{corpus_folder}/{partition}/annotations.csv", "w") as f:
            f.write("audio_file,label_id,label_name\n")
        
        annotations_file = f"{corpus_folder}/{partition}/annotations.csv"
        audios_dir = f"{corpus_folder}/{partition}/audios"

        for i, audio_data in enumerate(corpus_data[partition]):

            audio_file = f"{audios_dir}/audio_{partition}_{i}.wav"
            audio_array = resize_signal(audio_data["audio"]["array"])
            audio_label_id = audio_data[label_column]
            audio_label_name = label_mapper[audio_label_id]

            # Guardar anotación en fichero CSV
            store_annotation(audio_label_id, audio_label_name, audio_file, annotations_file)
            
            # Guardar el audio en formato WAV
            store_audio_file(audio_array, audio_file)

Finalmente, aplicamos la función anterior para obtener la estructura de directorios deseada:

generate_corpus_files(ccmusic_corpus, "fst_level_label", label_mapper_fst_level, "ccmusic")

generate_corpus_files(ccmusic_corpus, "sec_level_label", label_mapper_sec_level, "ccmusic2")

Observación: la celda anterior tarda bastante tiempo, debido al elevado tamaño del dataset (aproximadamente unos 20 minutos cada una), por lo que se recomienda no ejecutarla a menos que cuente con tiempo suficiente. Alternativamente, puede proceder a visualizar las salidas del presente notebook ya ejecutadas.

Vamos a visualizar la estructura de directorios ya creada:

def list_directory_structure(root_directory, max_depth=2):
    for root, _, files in os.walk(root_directory):
        level = root.replace(root_directory, '').count(os.sep)
        if level > max_depth:
            continue
        indent = ' ' * 4 * level
        print(f"{indent}{os.path.basename(root)}/")
        if level + 1 <= max_depth:
            subindent = ' ' * 4 * (level + 1)
            for f in files:
                print(f"{subindent}{f}")

list_directory_structure("ccmusic")

print()

list_directory_structure("ccmusic2")

ccmusic/
    test/
        annotations.csv
        features.csv
        audios/
    train/
        annotations.csv
        features.csv
        audios/
    validation/
        annotations.csv
        features.csv
        audios/

ccmusic2/
    test/
        annotations.csv
        audios/
    train/
        annotations.csv
        features.csv
        audios/
    validation/
        annotations.csv
        audios/

---